package com.hrvsr;

import java.io.Closeable;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hrvsr.common.Request;
import com.hrvsr.common.Response;
import com.hrvsr.common.Result;
import com.hrvsr.common.util.CollectionUtil;
import com.hrvsr.downloader.ClosableSafeDownloader;
import com.hrvsr.downloader.Downloader;
import com.hrvsr.downloader.httpclient.HttpClientDownloaderBuilder;
import com.hrvsr.observer.Observable;
import com.hrvsr.observer.Observer;
import com.hrvsr.observer.State;
import com.hrvsr.pipeline.ConsolePipeline;
import com.hrvsr.pipeline.Pipeline;
import com.hrvsr.processor.ResponseProcessor;
import com.hrvsr.scheduler.RequestFeeder;
import com.hrvsr.scheduler.Scheduler;
import com.hrvsr.scheduler.standardalone.StandardAloneScheduler;
import com.hrvsr.session.EmptySessionPool;
import com.hrvsr.session.Session;
import com.hrvsr.session.SessionPool;
import com.hrvsr.thread.BlockingCountableThreadPool;

public class Harvester implements Observable {
	private String nodeId;
	
	private static Logger logger = LoggerFactory.getLogger(Harvester.class);

	private int queueEmptySleepTime = 1000; // milliseconds
	private Scheduler scheduler;
	private SessionPool sessionPool;
	private Downloader downloader;
	private ResponseProcessor responseProcessor;
	private Pipeline pipeline;

	// -------request feeder----------
	private RequestFeeder requestFeeder;
	private AtomicBoolean feederAlive = new AtomicBoolean(true);

	// ---------observer-------------
	private Observer observer;
	private State state = new State(this.nodeId);

	// ---------thread and lock-----------
	private BlockingCountableThreadPool threadPool;
	private Lock lock = new ReentrantLock();
	private Condition canTake = lock.newCondition();

	private void stateChangedTo(int stat) {
		if (state.getStat() != stat) {
			state.setStat(stat);
			notifyObservers();
		}
	}

	public void run() {
		ensureComponents();
		stateChangedTo(State.RUNNING);
		feedRequests();
		while (true) {
			Request request = scheduler.poll();
			if (request == null) {
				if (meDone()) {// me done
					if (observer.clustersDone()) {
						break;
					} else {
						stateChangedTo(State.IDLE);
						waitOnCluster();
						continue;
					}
				} else {// me still alive
					waitOnMe();
					continue;
				}
			}
			stateChangedTo(State.RUNNING);
			Runnable mission = generateMission(request);
			threadPool.execute(mission);
		}
		close();
	}

	private void ensureComponents() {
		if (nodeId == null)
			throw new IllegalStateException("harvester node id cannot be null");

		if (state == null) {
			state = new State(nodeId);
			stateChangedTo(State.NOT_INITED);
		}

		if (scheduler == null) {
			scheduler = new StandardAloneScheduler();
		}

		if (sessionPool == null) {
			sessionPool = new EmptySessionPool();
		}

		if (downloader == null) {
			downloader = new HttpClientDownloaderBuilder().build();
		}
		downloader = ClosableSafeDownloader.adapt(downloader);

		if (responseProcessor == null) {
			
		}
		
		if(pipeline == null){
			pipeline = new ConsolePipeline();
		}
		
		stateChangedTo(State.INITED);
	}

	private void close() {
		threadPool.shutdown();
		// spin
		while (!meDone()) {
			try {
				Thread.sleep(200L);
			} catch (InterruptedException e) {
				logger.error("thread interrupted while trying to close harvester");
			}
		}

		destroy(scheduler);
		destroy(sessionPool);
		destroy(downloader);
		destroy(responseProcessor);
		destroy(pipeline);
		stateChangedTo(State.DONE);
		destroy(observer);
	}

	private void destroy(Object obj) {
		if (obj instanceof Closeable) {
			IOUtils.closeQuietly((Closeable) obj);
		}
	}

	private void feedRequests() {
		if (requestFeeder == null) {
			feederAlive.set(false);
			return;
		}

		Runnable r = new Runnable() {
			@Override
			public void run() {
				feederAlive.set(true);
				try {
					while (true) {
						Request request = requestFeeder.getNext();
						if (request == null) {
							break;
						}
						scheduler.offer(request);
						queueFilled();
						// Thread.yield();
					}
				} catch (Exception e) {
					logger.error("error feeding queue");
				} finally {
					feederAlive.set(false);
					destroy(requestFeeder);
				}
			}
		};

		Thread t = new Thread(r);
		t.start();
	}

	private void waitOnCluster() {
		try {
			TimeUnit.MICROSECONDS.sleep(queueEmptySleepTime);
		} catch (InterruptedException e) {
			logger.error("Harvester {} interruteped while waiting on cluster", nodeId);
		}
	}

	private void waitOnMe() {
		lock.lock();
		try {
			canTake.await(queueEmptySleepTime, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
			logger.error("Harvester {} interruteped while waiting on itself", nodeId);
		} finally {
			lock.unlock();
		}
	}

	public Runnable generateMission(final Request request) {
		return new Runnable() {
			@Override
			public void run() {
				Session session = sessionPool.poll(request);
				try {
					Response response = downloader.download(request, session);

					Result<?> result = responseProcessor.processResponse(response);

					// update status
					if (result.isSuccess()) {
						state.success();
					} else {
						state.failure();
					}
					chkAndAddRequests(result);

					if (result.needsPipeline()) {
						pipeline.pipeline(result);
					}
				} catch (Exception e) {
					logger.error("error doing mission for {}", request, e);
				} finally {
					sessionPool.offer(session);
					state.setDownloaderThreadNum(threadPool.getActiveCount());
					notifyObservers();
				}
			}
		};
	}

	private void chkAndAddRequests(Result<?> result) {
		if (result.needsRetry()) {
			scheduler.putEvenDuplicate(result.getRequest());
		} else {
			List<Request> nextRequests = result.getNextRequests();
			if (CollectionUtil.isEmpty(nextRequests))
				return;
			for (Request r : nextRequests) {
				scheduler.offer(r);
			}
		}
		// notify main thread to poll after put requests
		queueFilled();
	}

	@Override
	public void notifyObservers() {
		observer.update(state);
	}

	private void queueFilled() {
		lock.lock();
		try {
			canTake.signalAll();
		} finally {
			lock.unlock();
		}
	}

	public boolean meDone() {
		return threadPool.getActiveCount() <= 0 && !feederAlive.get();
	}

	// --------------getters ans setters-------------------
	public Scheduler getScheduler() {
		return scheduler;
	}

	public Harvester setScheduler(Scheduler scheduler) {
		this.scheduler = scheduler;
		return this;
	}

	public SessionPool getSessionPool() {
		return sessionPool;
	}

	public Harvester setSessionPool(SessionPool sessionPool) {
		this.sessionPool = sessionPool;
		return this;
	}

	public ResponseProcessor getResponseProcessor() {
		return responseProcessor;
	}

	public Harvester setResponseProcessor(ResponseProcessor responseProcessor) {
		this.responseProcessor = responseProcessor;
		return this;
	}

	public Pipeline getPipeline() {
		return pipeline;
	}

	public Harvester setPipeline(Pipeline pipeline) {
		this.pipeline = pipeline;
		return this;
	}

	public RequestFeeder getRequestFeeder() {
		return requestFeeder;
	}

	public Harvester setRequestFeeder(RequestFeeder requestFeeder) {
		this.requestFeeder = requestFeeder;
		return this;
	}

	public Observer getObserver() {
		return observer;
	}

	public Harvester setObserver(Observer observer) {
		this.observer = observer;
		return this;
	}

	public Downloader getDownloader() {
		return downloader;
	}

	public Harvester setDownloader(Downloader downloader) {
		this.downloader = downloader;
		return this;
	}

	public BlockingCountableThreadPool getThreadPool() {
		return threadPool;
	}

	public Harvester setThreadPool(BlockingCountableThreadPool threadPool) {
		this.threadPool = threadPool;
		return this;
	}

	public void setNodeId(String nodeId) {
		this.nodeId = nodeId;
		state = new State(nodeId);
	}

}